package belf.migrate.plugin.db2;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.exception.*;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.ExceptionUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

@Slf4j
public class DB2Catalog extends AbstractJdbcCatalog {

    private static final String QUOTE = "\"";

    static {
        SYS_DATABASES.addAll(Collections.unmodifiableList(
                Arrays.asList("SYSIBM", "SYSTOOLS")));
    }

    public DB2Catalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.DB_2, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getDatabaseAndTableName();
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    /**
     * 获取某个数据库或者模式下所有的表
     * @param database
     * @param schema
     * @return
     */
    @Override
    public String getListTableSQL(String database, String schema) {
        return String.format("""
                            SELECT TABNAME AS TABLE_NAME
                            FROM SYSCAT.TABLES
                            WHERE TABSCHEMA = '%s' AND TYPE = 'T'
                            ORDER BY TABNAME
                        """,
                schema.toUpperCase());
    }

    /**
     * 获取所有的数据库或者schema
     * @return
     */
    @Override
    public String getListDatabaseSQL() {
        return """
           SELECT
               TABSCHEMA
           FROM TABLE(ADMIN_GET_TAB_INFO('',''))
           GROUP BY TABSCHEMA
           """;
    }

    /**
     * 获取列表视图的SQL查询语句
     * 该方法旨在根据指定的模式名，生成一个SQL查询，用于获取该模式下所有视图的名称
     * 它通过连接系统目录表SYSCAT.TABLES和SYSCAT.VIEWS来识别视图，并根据视图的模式和类型进行过滤
     *
     * @param schemaName 模式名称，用于限定查询范围
     * @return 包含视图名称的SQL查询语句
     */
    @Override
    public String getListViewsSQL(String schemaName) {
        // refer: https://www.ibm.com/docs/en/db2/11.5?topic=views-syscatviews
        return String.format("""
                SELECT
                    VIEWNAME AS VIEW_NAME
                FROM
                    SYSCAT.VIEWS
                WHERE
                    VIEWSCHEMA = '%s';
                """, schemaName);
    }

    @Override
    public String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return getListViewsSQL(schemaName);
    }

    /**
     * 获取指定模式下的函数列表的SQL查询语句
     *
     * 此方法构造了一个SQL查询语句，用于从数据库的系统目录中获取指定模式下的所有函数名称
     * 它特别关注了函数的类型和模块状态，以确保只返回符合条件的函数
     *
     * @param schemaName 模式名称，用于限定查询范围
     * @return 返回一个SQL查询语句，用于获取指定模式下的函数列表
     */
    @Override
    public String getListFunctionSQL(String schemaName) {
        // refer: https://www.ibm.com/docs/en/db2/11.5?topic=views-syscatroutines
        return String.format("""
                SELECT
                    ROUTINENAME AS FUNCTION_NAME
                FROM
                    SYSCAT.ROUTINES
                WHERE
                    ROUTINESCHEMA = '%s'
                    AND ROUTINETYPE = 'F'
                    AND ORIGIN IN ('A','E','Q','U')                
                """, schemaName);
    }

    /**
     * 获取触发器列表的SQL语句
     *
     * 此方法根据给定的模式名称生成一个SQL查询语句，用于获取该模式下所有触发器的名称
     * 该SQL语句查询SYSCAT.TRIGGERS表，筛选出触发器所属模式与给定模式名称匹配的触发器，并按触发器名称排序
     *
     * @param schemaName 模式名称，用于筛选属于该模式的触发器
     * @return 返回生成的SQL语句字符串
     */
    @Override
    public String getListTriggerSQL(String schemaName) {
        // refer: https://www.ibm.com/docs/en/db2/11.5?topic=views-syscattriggers
        return String.format("""
                SELECT
                    TRIGNAME AS TRIGGER_NAME
                FROM
                    SYSCAT.TRIGGERS
                WHERE
                    TRIGSCHEMA = '%s';               
                """, schemaName);
    }

    /**
     * 获取指定模式下的索引列表SQL
     * 该方法用于动态生成SQL查询语句，旨在从数据库的系统目录中获取指定模式下所有索引的信息
     * 主要关注索引名（INDEX_NAME）和表名（TABLE_NAME），并对结果按照索引名进行排序
     *
     * @param schemaName 模式名称，用于过滤查询结果，只返回指定模式下的索引信息
     * @return 返回一个字符串，包含了一个可以直接执行的SQL查询语句
     */
    @Override
    public String getListIndexSQL(String schemaName) {
        // refer: https://www.ibm.com/docs/zh/db2/11.5?topic=views-syscatindexes
        return String.format("""
                SELECT
                    INDNAME AS INDEX_NAME,
                    TABNAME AS TABLE_NAME,
                    INDEXTYPE AS INDEX_TYPE
                FROM
                    SYSCAT.INDEXES
                WHERE
                    INDSCHEMA = '%s';
                """, schemaName);
    }


    /**
     * 获取数据库类型列表的SQL语句
     * 此方法用于构造一个SQL查询，该查询从指定的数据库中获取所有非系统数据类型
     * 它排除了系统数据类型（METATYPE <> 'S'），并按类型名称排序
     *
     * @param defaultDatabase 默认数据库名称，用于限定数据类型的schema
     * @return 返回一个SQL查询字符串，用于获取数据类型列表
     */
    @Override
    public String getListTypeSQL(String defaultDatabase) {
        // refer: https://www.ibm.com/docs/en/db2/11.5?topic=views-syscatdatatypes
        return String.format("""
                SELECT TYPENAME AS TYPE_NAME
                FROM SYSCAT.DATATYPES
                WHERE TYPESCHEMA = '%s'
                    AND METATYPE != 'S'
                """, defaultDatabase);
    }

    /**
     * 生成获取存储过程列表的SQL语句
     *
     * 此方法根据指定的数据库模式生成一个SQL查询语句，用于获取该数据库模式下所有存储过程的名称
     * 存储过程是数据库中一组为了完成特定功能而编写的SQL语句集合，它们可以接受参数、处理逻辑并返回结果
     *
     * @param defaultDatabase 数据库的默认模式，用于限定查询的范围
     * @return 用于获取存储过程列表的SQL查询语句
     */
    @Override
    public String getListProcedureSQL(String defaultDatabase) {
        // 使用String.format构建并返回SQL查询语句
        // 从SYSCAT.ROUTINES表中选择ROUTINENAME字段
        // WHERE子句用于过滤出存储过程类型（ROUTINETYPE='P'）且没有模块名称（ROUTINEMODULENAME IS NULL）的记录
        // 结果按ROUTINENAME排序
        return String.format("""
           SELECT
               ROUTINENAME
           FROM
               SYSCAT.ROUTINES
           WHERE
               ROUTINESCHEMA = '%s'
               AND ROUTINETYPE = 'P'
               AND ROUTINEMODULENAME IS NULL
           ORDER BY
               ROUTINENAME WITH UR
           """, defaultDatabase);
    }

    @Override
    public String getExistDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getTableName();
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getSchemaAndTableName();
    }


    /**
     * 列出指定数据库和模式下的所有表信息
     *
     * @param databaseName 数据库名称
     * @param schemaName   模式名称
     * @return 返回一个包含表信息的列表，每个元素代表一个表的信息模型
     * @throws CatalogException        当目录操作失败时抛出
     * @throws DatabaseNotExistException 当指定的数据库不存在时抛出
     */
    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName)
            throws CatalogException, DatabaseNotExistException {
        String sql = String.format("""
                SELECT
                    t.TABNAME AS TABLE_NAME,
                    t.CARD AS TABLE_ROWS,
                    t.NPAGES * ts.PAGESIZE AS TABLE_SIZE,
                    t.REMARKS AS TABLE_COMMENT
                FROM
                    SYSCAT.TABLES t JOIN SYSCAT.TABLESPACES ts ON t.TBSPACEID = ts.TBSPACEID
                WHERE
                    t.TABSCHEMA = '%s'
                    AND t.TYPE='T'
                """, schemaName);
        try (PreparedStatement ps =
                     getConnection(defaultUrl)
                             .prepareStatement(sql);
             ResultSet rs = ps.executeQuery()) {

            List<TableInfoModel> tables = new ArrayList<>();
            while (rs.next()) {
                TableInfoModel table = new TableInfoModel();
                table.setTableName(rs.getString("TABLE_NAME"));
                table.setRows(rs.getLong("TABLE_ROWS"));
                table.setSize(rs.getLong("TABLE_SIZE"));
                table.setComment(rs.getString("TABLE_COMMENT"));
                tables.add(table);
            }

            return tables;
        } catch (Exception e) {
            throw new CatalogException(
                    String.format("Failed listing table in catalog %s", catalogName), e);
        }
    }

    /**
     * 根据表路径获取表信息
     *
     * @param tablePath 表路径，包含 schema 名称和表名称
     * @return TableInfoModel 对象，包含表名、表大小和表行数
     * @throws CatalogException 如果获取表信息失败，则抛出此异常
     */
    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) throws CatalogException {
        TableInfoModel table = new TableInfoModel();
        try {
            String sql = String.format("""
                SELECT
                    t.TABNAME AS TABLE_NAME,
                    t.CARD AS TABLE_ROWS,
                    t.NPAGES * ts.PAGESIZE AS TABLE_SIZE,
                    t.REMARKS AS TABLE_COMMENT
                FROM
                    SYSCAT.TABLES t JOIN SYSCAT.TABLESPACES ts ON t.TBSPACEID = ts.TBSPACEID
                WHERE
                    t.TABSCHEMA = '%s'
                    AND t.TABNAME = '%s'
                    AND t.TYPE='T'
                """, tablePath.getSchemaName(), tablePath.getTableName());
            log.info("Query mysql table info: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setSize(resultSet.getLong("TABLE_SIZE"));
                table.setComment(resultSet.getString("TABLE_COMMENT"));
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("获取表的信息失败！", e);
        }
        return table;
    }

    /**
     * 获取序列列表的SQL语句
     *
     * 此方法用于构造一个SQL查询，该查询从DB2数据库的系统目录中检索指定模式下的所有序列对象
     * 序列是数据库中的一种对象，用于生成唯一的数值，通常用于主键或其它需要唯一标识的情况
     *
     * @param schema 模式名称，用于限定序列的范围
     * @return 返回一个SQL查询字符串，用于获取指定模式下的序列名称列表
     *
     * 注：此SQL查询特定于DB2数据库
     */
    @Override
    public String getListSequenceSQL(String schema) {
        // refer: https://www.ibm.com/docs/en/db2/11.5?topic=views-syscatsequences
        // SEQTYPE: I=Identity sequence，在同步表时会自动处理
        //          S=Sequence，表示通过CREATE SEQUENCE语句主动创建的序列，所以只需要返回S类型的SEQ即可
        return String.format("""
                SELECT SEQNAME AS SEQUENCE_NAME
                FROM SYSCAT.SEQUENCES
                WHERE SEQSCHEMA = '%s'
                    AND SEQTYPE ='S'
                """, schema);
    }

    /**
     * 获取列表同义词的SQL语句
     *
     * 此方法用于生成一个SQL查询，该查询从系统的目录视图中检索所有类型为'A'的表、序列和模块的名称，
     * 并将它们统一格式化为一个包含名称的列表这些对象被视为“同义词”，
     * 即它们可以互换使用，具有相同的意义和用途
     *
     * @param schema 架构名称，用于过滤属于特定架构的表、序列和模块
     * @return 返回一个格式化的SQL字符串，用于获取同义词列表
     */
    @Override
    public String getListSynonymSQL(String schema) {
        // refer:https://www.ibm.com/docs/zh/db2/11.5?topic=views-syscattables
        // TYPE = 'A'表示别名Alias(在DB2中等同于SYNONYM)
        // BASE_TABSCHEMA表示目标表的schema
        // BASE_TABNAME表示目标表的table_name
        return String.format("""
                SELECT
                    TABNAME AS SYNONYM_NAME
                FROM
                    SYSCAT.TABLES
                WHERE
                    TYPE = 'A' AND TABSCHEMA = '%s';
                """, schema);
    }

    @Override
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
        String sql = String.format("""
                SELECT
                  COLNAME AS COLUMN_NAME,
                  DEFAULT AS COLUMN_DEFAULT,
                  SCALE AS DATETIME_PRECISION,
                  REMARKS AS COLUMN_COMMENT
                FROM SYSCAT.COLUMNS
                WHERE TABSCHEMA ='%s'
                  AND TABNAME ='%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());

        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                String columnDefault = rs.getString("COLUMN_DEFAULT");
                int datetimePrecision = rs.getInt("DATETIME_PRECISION");
                String columnComment = rs.getString("COLUMN_COMMENT");
                Optional<ColumnMetaData> possibleColumn = columns.stream().filter(new Predicate<ColumnMetaData>() {
                    @Override
                    public boolean test(ColumnMetaData columnMetaData) {
                        return columnMetaData.getColumnName().equalsIgnoreCase(columnName);
                    }
                }).findFirst();
                if (possibleColumn.isPresent()) {
                    ColumnMetaData columnMetaData = possibleColumn.get();
                    ColumnExtra columnExtra = new ColumnExtra(columnDefault, null, columnComment, datetimePrecision);
                    columnMetaData.setExtra(columnExtra);
                }
            }
        } catch (SQLException e) {
            log.error("Error on query db2 column extra information:\n{}", ExceptionUtils.toString(e));
        }
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl = defaultUrl;

        // 查询指定表的索引，排除主键索引
        String sql = String.format("""
                SELECT
                    i.INDNAME AS INDEX_NAME,
                    i.INDEXTYPE AS INDEX_TYPE_NAME,
                    i.UNIQUERULE AS UNIQUENESS,
                    ic.COLNAME AS COLUMN_NAME,
                    ic.COLSEQ AS SEQ_IN_INDEX,
                    ic.COLORDER AS INDEX_ORDER
                FROM SYSCAT.INDEXES i
                    JOIN SYSCAT.INDEXCOLUSE ic ON i.INDNAME = ic.INDNAME
                WHERE
                    i.TABSCHEMA='%s'
                    AND i.TABNAME='%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());

        try {
            Connection conn = getConnection(dbUrl);
            Statement statement = conn.createStatement();

            DatabaseMetaData metaData = conn.getMetaData();
            List<IndexKey> defaultIndexKeys =  CatalogUtils.getIndexKeys(metaData, tablePath);

            // index name -> index
            Map<String, IndexKeyExt> indexKeyMap = new HashMap<>();

            try (ResultSet rs = statement.executeQuery(sql)) {
                log.debug("query index:{}", sql);
                while (rs.next()) {
                    boolean unique = "U".equals(rs.getString("UNIQUENESS"))?true:false;
                    String indexName = rs.getString("INDEX_NAME");
                    int sequenceInIndex = rs.getInt("SEQ_IN_INDEX");
                    String columnName = rs.getString("COLUMN_NAME");
                    String indexOrder = rs.getString("INDEX_ORDER");
                    String indexTypeName = rs.getString("INDEX_TYPE_NAME");
                    List<IndexKeyExt.ColumnWithSequence> columns = new ArrayList<>();
                    columns.add(new IndexKeyExt.ColumnWithSequence(columnName, sequenceInIndex));
                    IndexKeyExt IndexKeyExt = new IndexKeyExt(
                            columns,
                            indexName,
                            unique,
                            indexOrder,
                            indexTypeName);
                    IndexKeyExt existOne = indexKeyMap.get(indexName);
                    if (existOne != null) {
                        existOne.getColumns().addAll(IndexKeyExt.getColumns());
                    } else {
                        indexKeyMap.put(indexName, IndexKeyExt);
                    }

                    log.debug("got index: {}", IndexKeyExt);
                }
            }

            Iterator<IndexKey> it = defaultIndexKeys.iterator();
            while (it.hasNext()) {
                IndexKey indexKey = it.next();
                IndexKeyExt IndexKeyExt = indexKeyMap.get(indexKey.getIndexName());
                if (null != IndexKeyExt) {
                    indexKey.setColumnNames(IndexKeyExt.getSortedColumnNames());
                    indexKey.setIndexTypeName(IndexKeyExt.getIndexTypeName());
                    if (IndexKeyExt.getIndexOrder() != null
                            && "ASC".equalsIgnoreCase(IndexKeyExt.getIndexOrder())) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.ASC);
                    } else if (IndexKeyExt.getIndexOrder() != null
                            && "DESC".equalsIgnoreCase(IndexKeyExt.getIndexOrder())) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.DESC);
                    }
                } else { // DB2自身查找索引不包含主键和Function-based Normal索引，把这些索引相应删除了
                    it.remove();
                }
            }

            statement.close();
            return defaultIndexKeys;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 显示db2 创建表的语句。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        String sql = null;
        try {
            String queryOpToken = String.format("CALL SYSPROC.DB2LK_GENERATE_DDL('-e -x -xd -td -t %s.%s',?)", tablePath.getSchemaName(), tablePath.getTableName());
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(queryOpToken);
            int token = resultSet.getInt(1);
            sql = String.format("SELECT SQL_STMT FROM SYSTOOLS.DB2LOOK_INFO WHERE OP_TOKEN = %s ORDER BY OP_SEQUENCE WITH UR", token);
            resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("SQL_STMT");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new TableNotExistException(DatabaseIdentifier.ORACLE, tablePath);
        }
        return null;
    }

    /**
     * 显示db2 创建视图的语句。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {
        String sql = String.format("""
                SELECT TEXT AS VIEW_DDL
                FROM SYSCAT.VIEWS
                WHERE VIEWSCHEMA ='%s' AND VIEWNAME='%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("VIEW_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new ViewNotExistException(DatabaseIdentifier.DB_2, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        String query = String.format("""
                        SELECT TEXT AS PROCEDURE_DDL
                        FROM SYSCAT.ROUTINES
                        WHERE ROUTINESCHEMA ='%s' AND ROUTINENAME='%s'
                        AND ROUTINETYPE ='P'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("PROCEDURE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create procedure : \n{}", sql);
            throw new ProcedureNotExistException(DatabaseIdentifier.DB_2, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException{
        String sql = String.format("""
                SELECT TEXT AS TRIGGER_DDL
                FROM SYSCAT.TRIGGERS
                WHERE TRIGSCHEMA ='%s' AND TRIGNAME='%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("TRIGGER_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create trigger: \n{}", sql);
            throw new TriggerNotExistException(DatabaseIdentifier.DB_2, tablePath, e);
        }
        return null;
    }

    /**
     * DB2中没有直接显示自定义类型对象DDL的SQL，重构自定义类型对象的DDL非常麻烦，以后再说。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateType(TablePath tablePath) throws TypeNotExistException {
        // refer: https://www.ibm.com/docs/zh/db2/11.5?topic=cv-syscatdatatypes
        // refer: https://www.ibm.com/docs/zh/db2/11.5?topic=views-syscatattributes
        // 从SYSCAT.DATATYPES查到UFT名称和类型，从SYSCAT.ATTRIBUTES查UDT属性，在拼接DDL
        // TODO: 等待完善
        return null;
    }

    @Override
    public String showCreateSeq(TablePath tablePath) throws SequenceNotExistException {
        try {
            Connection connection = getConnection(defaultUrl);
            return new DB2DDLBuilder().getSequenceDDL(connection, tablePath);
        } catch (Exception e) {
            log.error("Error while show create sequence: \n{}", tablePath.getTableName());
            throw new SequenceNotExistException(DatabaseIdentifier.DB_2, tablePath, e);
        }
    }

    @Override
    public String showCreateSynonym(TablePath tablePath) {
        String ddl = "";
        try {
            Connection connection = getConnection(defaultUrl);
            ddl = new DB2DDLBuilder().getSynonymDDL(connection, tablePath);
        } catch (Exception e) {
            log.error("Error while show create sequence: \n{}", tablePath.getTableName());
        }
        return ddl;
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        String ddl = "";
        String sql = String.format("""
                SELECT TEXT AS FUNC_DDL FROM SYSCAT.ROUTINES
                WHERE
                    ROUTINESCHEMA = '%s'
                    AND ROUTINENAME = '%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                ddl = resultSet.getString("FUNC_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create function: \n{}", sql);
            throw new FunctionNotExistException(DatabaseIdentifier.DB_2, tablePath, e);
        }
        return ddl;
    }

    @Override
    public String showCreateIndex(TablePath tablePath) throws IndexNotExistException {
        String sql = String.format("""
                WITH INDEX_COLUMNS AS (
                    SELECT
                        INDNAME,
                        COLNAME,
                        COLSEQ,
                        COLORDER
                    FROM
                        SYSCAT.INDEXCOLUSE
                    WHERE
                        INDNAME = '%s'
                        AND INDSCHEMA = '%s'
                ),
                INDEX_INFO AS (
                    SELECT
                        INDNAME,
                        TABNAME,
                        UNIQUERULE,
                        INDEXTYPE,
                        COLNAMES
                    FROM
                        SYSCAT.INDEXES
                    WHERE
                        INDNAME = '%s'
                        AND INDSCHEMA = '%s'
                )
                SELECT
                    CASE WHEN ROW_NUMBER() OVER(PARTITION BY I.INDNAME ORDER BY IC.COLSEQ) = 1 THEN
                        'CREATE ' ||
                        CASE WHEN I.UNIQUERULE = 'U' THEN 'UNIQUE '
                             ELSE ''
                        END ||
                        'INDEX ' || I.INDNAME || ' ON ' || I.TABNAME || '('
                    ELSE
                        ''
                    END ||
                    IC.COLNAME ||
                    CASE WHEN IC.COLORDER = 'D' THEN ' DESC' ELSE '' END ||
                    CASE WHEN LAST_VALUE(IC.COLSEQ) OVER(PARTITION BY I.INDNAME ORDER BY IC.COLSEQ ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) = IC.COLSEQ THEN ')'
                    ELSE ','
                    END AS INDEX_DDL
                FROM
                    INDEX_INFO I
                JOIN
                    INDEX_COLUMNS IC ON I.INDNAME = IC.INDNAME
                ORDER BY
                    IC.COLSEQ
                """,
                tablePath.getTableName(),
                tablePath.getSchemaName(),
                tablePath.getTableName(),
                tablePath.getSchemaName());

        try {
            return getDDLs(sql, "INDEX_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create index: \n{}", sql);
            throw new IndexNotExistException(DatabaseIdentifier.DB_2, tablePath, e);
        }
    }

    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
                SELECT
                    TC.CONSTNAME AS CONSTRAINT_NAME,
                    TC.TABNAME AS TABLE_NAME,
                    K.COLNAME AS COLUMN_NAME,
                    TC.TYPE AS CONSTRAINT_TYPE
                FROM
                    SYSCAT.TABCONST TC
                JOIN
                    SYSCAT.KEYCOLUSE K ON TC.CONSTNAME = K.CONSTNAME AND TC.TABNAME = K.TABNAME
                WHERE
                    TC.TABSCHEMA ='%s'
                    AND TC.TABNAME = '%s'
                    AND TC.TYPE = 'U' -- U for unique constraint
                ORDER BY
                    TC.CONSTNAME, K.COLSEQ;
                    """, tablePath.getSchemaName(), tablePath.getTableName());
        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String column = rs.getString("COLUMN_NAME");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "DB2查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
                SELECT
                    TC.TABSCHEMA AS TABLE_SCHEMA,
                    TC.TABNAME AS TABLE_NAME,
                    TC.CONSTNAME AS CONSTRAINT_NAME,
                    C.TYPE AS CONSTRAINT_TYPE,
                    C.TEXT AS CHECK_CONDITION
                FROM
                    SYSCAT.TABCONST TC
                JOIN
                    SYSCAT.CHECKS C ON TC.CONSTNAME = C.CONSTNAME AND TC.TABNAME = C.TABNAME
                WHERE
                    C.TYPE = 'C' -- C FOR CHECK CONSTRAINT
                    AND TC.TABSCHEMA='%s'
                    AND TC.TABNAME='%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String condition = rs.getString("CHECK_CONDITION");
                    checkConstraints.add(new CheckConstraint(tablePath.getTableName(),
                            constraintName,
                            condition));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Check约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }

    @Override
    public List<UserDefSeq> listSequences(String schema, List<String> sequenceNames) throws CatalogException {

        List<UserDefSeq> userDefSeqs = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);
        Map<String, String> seqDDLs = new DB2DDLBuilder().buildSequenceDDL(connection, schema, sequenceNames);
        for (Map.Entry<String, String> entry : seqDDLs.entrySet()) {
            userDefSeqs.add(new UserDefSeq(entry.getKey(), entry.getValue()));
        }
        return userDefSeqs;
    }

    @Override
    public List<UserDefFunction> listFunctions(String schema, List<String> functionNames) throws CatalogException {
        String sql = String.format("""
                SELECT 
                    ROUTINENAME,TEXT,ROUTINETYPE  
                FROM 
                    SYSCAT.ROUTINES 
                WHERE ROUTINESCHEMA='%s'
                    AND ROUTINETYPE='F'
                    AND ORIGIN IN ('A', 'E', 'Q', 'U')
                """, schema);
        if (null != functionNames && !functionNames.isEmpty()) {
            List<String> quotedFuncNames = new ArrayList<>();
            for (String funcName: functionNames) {
                quotedFuncNames.add("'" + funcName + "'");
            }
            sql += String.format(" AND ROUTINENAME IN (%s)", String.join(",", quotedFuncNames));
        }
        List<UserDefFunction> userDefFunctions = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefFunction userDefFunction = new UserDefFunction();
                    String functionName = rs.getString("ROUTINENAME");
                    userDefFunction.setFunctionName(functionName);
                    userDefFunction.setDdl(rs.getString("TEXT"));
                    userDefFunctions.add(userDefFunction);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_LIST_ERROR, "查询所有自定义函数的DDL失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefFunctions;
    }

    @Override
    public List<UserDefSynonym> listSynonyms(String schema, List<String> synonymNames) throws CatalogException {
        String sql = String.format("""
                SELECT TABNAME 
                FROM SYSCAT.TABLES 
                WHERE TYPE = 'A' AND TABSCHEMA = '%s';
                """, schema);
        if (null != synonymNames && !synonymNames.isEmpty()) {
            List<String> quotedFuncNames = new ArrayList<>();
            for (String synonymName: synonymNames) {
                quotedFuncNames.add("'" + synonymName + "'");
            }
            sql += String.format(" WHERE TABNAME IN (%s)", String.join(",", quotedFuncNames));
        }
        List<UserDefSynonym> userDefSynonyms = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefSynonym userDefSynonym = new UserDefSynonym();
                    String synonymName = rs.getString("TABNAME");
                    String ddl = showCreateSynonym(TablePath.of(null, schema, synonymName, null));
                    userDefSynonym.setDdl(ddl);
                    userDefSynonyms.add(userDefSynonym);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_LIST_ERROR, "查询所有自定义同义词的DDL失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefSynonyms;
    }

    @Override
    protected String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + tablePath.getSchemaAndTableName();
    }

    @Override
    public String getUrlFromDatabaseName(String database) {
        return defaultUrl;
    }

}
